package com.juc;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author eason
 * @description CyclicBarrier 使用demo
 * @date
 */
public class CyclicBarrierTest {
    private static Integer runs = 3;
    private static CyclicBarrier barrier = new CyclicBarrier(runs);

    public static void main(String[] args) {
        // 加载数据
        List<Integer> list = new ArrayList<>();
        for (int i = 0; i < 14; i++) {
            list.add(i);
        }

        // 总待处理的数量
        int sum = list.size();
        // 剩余待处理数量
        int remain = sum % runs;

        int loop = sum / runs;
        System.out.println("轮询的总次数:" + (loop + (remain > 0 ? 1 : 0)));
        try {
            // 线程池 保证corePoolSize>=runs
            final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(3, 8, 10, TimeUnit.MILLISECONDS,
                    new ArrayBlockingQueue<Runnable>(500), new ThreadPoolExecutor.DiscardPolicy());

            // 处理次数
            int p = 0;
            for (int i = 0; i < loop; i++) {
                for (int j = 0; j < runs; j++) {
                    threadPool.execute(new WorkThread(barrier, list.get(p)));
                    p++;
                }
            }

            // 剩余待处理
            if (remain > 0) {
                barrier = new CyclicBarrier(remain);
                for (int i = 0; i < remain; i++) {
                    threadPool.execute(new WorkThread(barrier, list.get(p)));
                    p++;
                }
            }

            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println("监听无任务时关闭线程池");
                        for (; ; ) {
                            Thread.sleep(1000);
                            //System.out.println("活跃的线程数："+ threadPool.getActiveCount() + ",核心线程数：" + threadPool.getCorePoolSize() + ",线程池大小：" + threadPool.getPoolSize() + ",队列的大小" + threadPool.getQueue().size());
                            if (threadPool.getActiveCount() == 0) {
                                System.out.println("关闭线程池");
                                threadPool.shutdown();
                                break;
                            }
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();

            System.out.println("主线程结束");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

class WorkThread implements Runnable {

    private Integer i;
    CyclicBarrier barrier;

    public WorkThread(CyclicBarrier barrier, Integer i) {
        this.barrier = barrier;
        this.i = i;
    }

    @Override
    public void run() {
        try {
            // 产生1000到3000之间的随机整数
            int randomNum = new Random().nextInt((3000 - 1000) + 1) + 1000;
            Thread.sleep(randomNum);
            System.out.println(Thread.currentThread().getName() + " 处理:" + i + ", 用时:" + ((double) randomNum / 1000) + "s");
            int arrival_index = barrier.await();
            if (0 == arrival_index) {
                System.out.println("First arrival thread in this iteration is: " + Thread.currentThread().getName());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

